package com.torry.edss.application.biz.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.torry.boot.common.util.Preconditions;
import com.torry.edss.application.biz.SyncDataBiz;
import com.torry.edss.application.cache.TradingDayCache;
import com.torry.edss.application.exception.ReplayExceptionEnum;
import com.torry.edss.domain.accessor.*;
import com.torry.edss.domain.model.*;
import com.torry.edss.domain.valueobject.MarketPlate;
import com.torry.edss.domain.valueobject.QuantityType;
import com.torry.edss.infras.entity.CompanyDO;
import com.torry.edss.infras.entity.HisDailyLineDO;
import com.torry.edss.infras.entity.LimitUpDO;
import com.torry.edss.infras.util.CompanyUtil;
import com.torry.edss.infras.util.DateUtil;
import com.torry.edss.infras.util.NumberUtils;
import com.torry.edss.integration.facade.RemoteApiFacade;
import com.torry.edss.integration.facade.enums.Direction;
import com.torry.edss.integration.facade.result.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author torry.wang
 */
@Service
@Slf4j
public class SyncDataBizImpl implements SyncDataBiz {

    @Resource
    private RemoteApiFacade remoteApiFacade;
    @Resource
    private CompanyAccessor companyAccessor;
    @Resource
    private LimitUpAccessor limitUpAccessor;
    @Resource
    private LimitBombAccessor limitBombAccessor;
    @Resource
    private LimitDownAccessor limitDownAccessor;
    @Resource
    private MarketOverviewAccessor marketOverviewAccessor;
    @Resource
    private HisDailyLineAccessor hisDailyLineAccessor;


    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean syncAllCompanies() {
        List<FetchAllStocksResult> stocks = remoteApiFacade.fetchAllStocks();
        List<Company> exists = companyAccessor.list();
        log.info("fetch companies size ：{}, existed companies size: {}", stocks.size(), exists.size());

        Set<String> existCodes = exists.stream().map(Company::getCode).collect(Collectors.toSet());
        Set<String> currentCodes = stocks.stream().map(FetchAllStocksResult::getCode).collect(Collectors.toSet());
        List<String> needlessCodes = existCodes.stream().filter(e -> !currentCodes.contains(e)).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(needlessCodes)) {
            // 已经退市
            companyAccessor.update(new LambdaUpdateWrapper<CompanyDO>().set(CompanyDO::getDelisted, 1).in(CompanyDO::getCode, needlessCodes));
        }
        Map<String, Company> codeToCompanyMap = exists.stream().collect(Collectors.toMap(Company::getCode, Function.identity()));

        // 过滤掉无变化的
        stocks = stocks.stream().filter(e -> {
            if (codeToCompanyMap.containsKey(e.getCode())) {
                Company oldCompanyInfo = codeToCompanyMap.get(e.getCode());
                if (oldCompanyInfo.getLdate() != null && oldCompanyInfo.getName().equals(e.getName())) {
                    return false;
                }
            }
            return true;
        }).collect(Collectors.toList());

        List<Company> companies = stocks.stream().map(companyResult -> {

            Company company = new Company();
            company.setName(companyResult.getName());
            company.setCode(companyResult.getCode());
            company.setBourse(companyResult.getBourse());
            company.setPlate(MarketPlate.getMarketPlate(companyResult.getCode()));
            company.setAcronym(CompanyUtil.parseCompanyAcronym(company.getName()));
            if (codeToCompanyMap.containsKey(companyResult.getCode())) {
                Company oldCompanyInfo = codeToCompanyMap.get(companyResult.getCode());
                company.setId(oldCompanyInfo.getId());
                company.setLdate(oldCompanyInfo.getLdate());
                company.setDelisted(oldCompanyInfo.getDelisted());
            } else {
                company.setDelisted(false);
                log.info(" insert new company: {}, {}", company.getName(), company.getCode());
            }
            if (Objects.isNull(company.getLdate())) {
                company.setLdate(remoteApiFacade.fetchStockLDate(company.getCode()));
            }
            // TODO 同步公司最新市值，因为调用较为频繁，先不实现。
            return company;
        }).collect(Collectors.toList());
        return companyAccessor.saveOrUpdateBatch(companies);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean syncLimitUps(String date) {
        if (!TradingDayCache.isTradingDay(date)) return true;
        if (limitUpAccessor.countByDate(date) > 0) return true;

        List<FetchLimitUpResult> limitUps = remoteApiFacade.fetchLimitUps(date);

        // 更新前一天涨停股票的断板信息
        List<LimitUp> previousLimitUps = limitUpAccessor.findByDate(TradingDayCache.getPreviousTradingDay());
        previousLimitUps.forEach(e -> {
            Long companyId = e.getCompanyId();
            Company company = companyAccessor.getById(companyId);
            e.setInterrupted(limitUps.stream().noneMatch(item -> item.getCode().equals(company.getCode())));
        });
        if (CollectionUtils.isNotEmpty(previousLimitUps)) {
            limitUpAccessor.updateBatchById(previousLimitUps);
        }

        List<LimitUp> models = limitUps.stream().map(e -> {
            LimitUp limitUp = new LimitUp();
            limitUp.setDate(DateUtil.parseDate(date));
            Long companyId = companyAccessor.getOne(new LambdaQueryWrapper<CompanyDO>().eq(CompanyDO::getCode, e.getCode())).getId();
            limitUp.setCompanyId(companyId);
            limitUp.setPrice(e.getPrice());
            limitUp.setUpPercentage(e.getUpPercentage());
            limitUp.setMarketValue(e.getLtsz());
            limitUp.setTransactionAmount(e.getAmount());
            limitUp.setSealAmount(e.getSealAmount());
            limitUp.setTurnoverPercentage(e.getTurnoverRate());
            limitUp.setLimitTime(LocalTime.parse(e.getFirstLimitUpTime(), DateTimeFormatter.ofPattern("HH:mm:ss")));
            limitUp.setLastLimitTime(LocalTime.parse(e.getLastLimitUpTime(), DateTimeFormatter.ofPattern("HH:mm:ss")));
            limitUp.setBombCount(e.getBombCount());
            limitUp.setStat(e.getStat());
            limitUp.setContinuousCount(e.getContinuousCount());
            limitUp.setSealTransactionPercentage(NumberUtils.format(100D * e.getSealAmount() / e.getAmount()));

            // 是否反包涨停
            if (e.getContinuousCount() > 1) {
                limitUp.setCovered(false);
            } else {
                int index = TradingDayCache.getAllTradingDays().indexOf(date);
                String beforeDay = TradingDayCache.getAllTradingDays().get(index - 2);
                int count = limitUpAccessor.countByDateAndCompanyId(Collections.singletonList(beforeDay), companyId);
                limitUp.setCovered(count > 0);
            }

            // 与前一次涨停间隔, 大于30天的全部默认为99
            if (e.getContinuousCount() > 1) {
                limitUp.setPreviousLimitUpIntervalDays(0);
            } else if (limitUp.getCovered()) {
                limitUp.setPreviousLimitUpIntervalDays(1);
            } else {
                List<LimitUp> list = limitUpAccessor.list(new LambdaQueryWrapper<LimitUpDO>()
                        .lt(LimitUpDO::getDate, date)
                        .eq(LimitUpDO::getCompanyId, companyId)
                        .orderByDesc(LimitUpDO::getDate));
                if (CollectionUtils.isEmpty(list)) {
                    limitUp.setPreviousLimitUpIntervalDays(99);
                } else {
                    int index = TradingDayCache.getAllTradingDays().indexOf(date);
                    LocalDate lastDate = list.get(0).getDate();
                    int intervalDays = index - TradingDayCache.getAllTradingDays().indexOf(DateUtil.formatDate(lastDate)) + 1;
                    limitUp.setPreviousLimitUpIntervalDays(intervalDays > 30 ? 99 : intervalDays);
                }
            }

            // 是否断板
            if (!TradingDayCache.getLastTradingDay().equals(date)) {
                int index = TradingDayCache.getAllTradingDays().indexOf(date);
                String nextDay = TradingDayCache.getAllTradingDays().get(index + 1);
                int count = limitUpAccessor.countByDateAndCompanyId(Collections.singletonList(nextDay), companyId);
                limitUp.setInterrupted(count == 0);
            }

            List<FetchDailyLineResult> allDailyData = remoteApiFacade.fetchAllDailyLines(e.getCode());

            int index = 0;
            for (int i = 0; i < allDailyData.size(); i++) {
                if (allDailyData.get(i).getDate().equals(date)) {
                    index = i;
                    break;
                }
            }
            // 涨停当日量能
            FetchDailyLineResult limitDayTradeInfo = allDailyData.get(index);
            FetchDailyLineResult limitPreviousDayTradeInfo = allDailyData.get(index - 1);
            Double v1 = limitDayTradeInfo.getVolume();
            // 涨停前一日量能
            Double v2 = limitPreviousDayTradeInfo.getVolume();
            limitUp.setQuantityType(v1 > v2 ? QuantityType.MAGNIFY : v1 < v2 ? QuantityType.SHRINK : QuantityType.EQUALITY);
            limitUp.setQuantity(v1);
            limitUp.setPreviousDayQuantity(v2);
//            limitUp.setOpenUpPercentage(NumberUtils.format(100D * (limitDayTradeInfo.getO() - limitPreviousDayTradeInfo.getC()) / limitPreviousDayTradeInfo.getC()));
            return limitUp;
        }).collect(Collectors.toList());

        if (CollectionUtils.isNotEmpty(models)) {
            limitUpAccessor.saveBatch(models);
        }
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean syncLimitBombs(String date) {
        if (!TradingDayCache.isTradingDay(date)) return true;
        if (limitBombAccessor.countByDate(date) > 0) return true;

        List<FetchLimitBombResult> bombs = remoteApiFacade.fetchLimitBombs(date);
        bombs = bombs.stream().filter(e -> !e.getCode().startsWith("8")).collect(Collectors.toList());
        List<LimitBomb> list = bombs.stream().map(e -> {
            Company company = companyAccessor.getByCode(e.getCode());
            if (company == null) return null;
            LimitBomb limitBomb = new LimitBomb();
            limitBomb.setDate(DateUtil.parseDate(date));
            limitBomb.setCompanyId(company.getId());
            limitBomb.setPrice(e.getPrice());
            limitBomb.setLimitPrice(e.getLimitUpPrice());
            limitBomb.setUpPercentage(e.getUpPercentage());
            limitBomb.setTransactionAmount(e.getAmount());
            limitBomb.setMarketValue(e.getLtsz());
            limitBomb.setTurnoverPercentage(e.getTurnoverRate());
            limitBomb.setStat(e.getStat());
            limitBomb.setFirstLimitTime(LocalTime.parse(e.getFirstLimitTime(), DateTimeFormatter.ofPattern("HH:mm:ss")));
            limitBomb.setBombCount(e.getBombCount());

            FetchDailyLineResult tradeInfo = remoteApiFacade.fetchDayLine(company.getCode(), date, Direction.CURRENT);

            // 当日及前一K线成交量
            if (tradeInfo != null) {
                limitBomb.setQuantity(tradeInfo.getVolume());
                FetchDailyLineResult previousKLineData = remoteApiFacade.fetchPreviousDayLine(company.getCode(), date);
                limitBomb.setPreviousDayQuantity(previousKLineData != null ? previousKLineData.getVolume() : null);
            }
            return limitBomb;
        }).filter(Objects::nonNull).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(list)) {
            limitBombAccessor.saveBatch(list);
        }

        //更新前一日炸板股池 第二天的开盘情况
        List<LimitBomb> yesterdayLimitBombs = limitBombAccessor.findByDate(TradingDayCache.getPreviousTradingDay(date));
        yesterdayLimitBombs.forEach(e -> {
            Long companyId = e.getCompanyId();
            Company company = companyAccessor.getById(companyId);
            FetchDailyLineResult tradeInfo = remoteApiFacade.fetchDayLine(company.getCode(), date, Direction.CURRENT);
            if (tradeInfo != null) {
                e.setNextDayOpenUpPercentage(NumberUtils.format(100D * (tradeInfo.getOpen() - e.getPrice().doubleValue()) / e.getPrice().doubleValue()));
                limitBombAccessor.updateById(e);
            }
        });
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean syncLimitDowns(String date) {
        if (!TradingDayCache.isTradingDay(date)) return true;
        if (limitDownAccessor.countByDate(date) > 0) return true;

        List<FetchLimitDownResult> limitDowns = remoteApiFacade.fetchLimitDowns(date);
        limitDowns = limitDowns.stream().filter(e -> StringUtils.isNotBlank(e.getCode().replaceAll(" ", ""))).collect(Collectors.toList());
        List<LimitDown> list = limitDowns.stream().map(e -> {
            LimitDown limitDown = new LimitDown();
            Company company = companyAccessor.getByCode(e.getCode());

            FetchDailyLineResult tradeInfo = remoteApiFacade.fetchDayLine(company.getCode(), date, Direction.CURRENT);

            // 当日及前一K线成交量
            if (tradeInfo != null) {
                limitDown.setQuantity(tradeInfo.getVolume());
                FetchDailyLineResult previousKLineData = remoteApiFacade.fetchPreviousDayLine(company.getCode(), date);
                limitDown.setPreviousDayQuantity(previousKLineData != null ? previousKLineData.getVolume() : null);
            }
            limitDown.setCompanyId(company.getId());
            limitDown.setDate(DateUtil.parseDate(date));
            limitDown.setPrice(e.getPrice());
            limitDown.setUpPercentage(e.getUpPercentage());
            limitDown.setTransactionAmount(e.getAmount());
            limitDown.setMarketValue(e.getLtsz());
            limitDown.setTurnoverPercentage(e.getTurnoverRate());
            limitDown.setContinuousCount(e.getContinuousCount());
            limitDown.setSealAmount(e.getSealAmount());
            limitDown.setMinPriceTransactionAmount(e.getMinPriceAmount());
            limitDown.setBombCount(e.getBombCount());
            return limitDown;
        }).collect(Collectors.toList());
        if (CollectionUtils.isNotEmpty(list)) {
            limitDownAccessor.saveBatch(list);
        }
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean syncMarketSituation(String date) {
        if (!TradingDayCache.isTradingDay(date)) return true;

        MarketOverview exponent = marketOverviewAccessor.getByDate(date);
        if (exponent == null) {
            exponent = new MarketOverview();
        }

        if (TradingDayCache.getLastTradingDay().equals(date)) {
            // 主力资金
            FetchFundFlowInfoResult fetchFundFlowInfoResult = remoteApiFacade.fetchFundFlowInfo(date);
            exponent.setDomesticCapital(NumberUtils.format(fetchFundFlowInfoResult.getTotalAmount() / 100000000));

            FetchRealTimeMarketDataResult.IndexQuote indexQuote = remoteApiFacade.fetchRealTimeMarketData().getIndexQuotes()
                    .stream().filter(e -> e.getCode().equals("sh000001")).findFirst().orElse(null);
            Preconditions.ofNotNull(indexQuote, ReplayExceptionEnum.CAN_NOT_FOUND_CURRENT_DAY_TRADE_INFO, "sh000001", date);
            exponent.setPrice(BigDecimal.valueOf(indexQuote.getPrice()));
            exponent.setUpPercentage(indexQuote.getUpPercentage());
            exponent.setUpValue(indexQuote.getUpValue());
            try {
                MarketOverview previousExponent = marketOverviewAccessor.getByDate(TradingDayCache.getPreviousTradingDay());
                Double realTimeMarketTradeAmount = remoteApiFacade.fetchRealTimeTotalAmount(previousExponent.getTransactionAmount());
                exponent.setTransactionAmount(realTimeMarketTradeAmount);
            } catch (Exception e) {
                log.info("获取两市成交额 error");
            }

            FetchUpDownInfoResult fetchUpDownInfoResult = remoteApiFacade.fetchRealTimeUpDownInfo();
            exponent.setUpCount(fetchUpDownInfoResult.getRiseNum());
            exponent.setDownCount(fetchUpDownInfoResult.getFallNum());
            exponent.setFixedCount(fetchUpDownInfoResult.getFlatNum());

        } else {
            // TODO 同步历史大盘数据， 之前使用的是mairui api，现在暂时没有相关接口
            System.out.println("TODO ");
        }

        List<FetchLimitUpResult> limitUps = remoteApiFacade.fetchLimitUps(date);
        List<FetchLimitDownResult> limitDowns = remoteApiFacade.fetchLimitDowns(date);

        // 炸板数
        int bombsCount = remoteApiFacade.fetchLimitBombs(date).size();

        // 连板数
        long lbc = limitUps.stream().filter(e -> e.getContinuousCount() > 1).count();
        List<LimitUp> previousDayLimitUps = limitUpAccessor.findByDate(TradingDayCache.getPreviousTradingDay(date));

        // 昨涨停 今天收盘有溢价的 数量
        int yjc = 0;
        for (LimitUp previousDayLimitUp : previousDayLimitUps) {
            Company company = companyAccessor.getById(previousDayLimitUp.getCompanyId());
            FetchDailyLineResult KLineData = remoteApiFacade.fetchDayLine(company.getCode(), date, Direction.CURRENT);
            yjc += KLineData == null ? 0 : KLineData.getPrice() > previousDayLimitUp.getPrice().doubleValue() ? 1 : 0;
        }

        // 昨涨停 今跌停的数量
        long hanCount = previousDayLimitUps.stream().filter(yesterdayLimitUp ->
                limitDowns.stream().map(FetchLimitDownResult::getCode).collect(Collectors.toList())
                        .contains(companyAccessor.getById(yesterdayLimitUp.getCompanyId()).getCode())
        ).count();

        exponent.setDate(DateUtil.parseDate(date));
        exponent.setLimitUpCount(limitUps.size());
        exponent.setLimitDownCount(limitDowns.size());
        // 封板率 涨停 / (涨停 + 炸板)
        exponent.setSealPercentage(NumberUtils.format(100D * limitUps.size() / (limitUps.size() + bombsCount)));
        // 连板率  连板数 / 昨涨停 数
        exponent.setContinuousPercentage(NumberUtils.format(100D * lbc / previousDayLimitUps.size()));
        // 溢价率  昨涨停今有溢价数 / 昨涨停数
        exponent.setPremiumPercentage(NumberUtils.format(100D * yjc / previousDayLimitUps.size()));
        // 核按钮  昨涨停今跌停数 / 昨涨停数
        exponent.setNuclearPercentage(NumberUtils.format(100D * hanCount / previousDayLimitUps.size()));

        // 最高连板数
        Optional<FetchLimitUpResult> max = limitUps.stream().max(Comparator.comparingInt(FetchLimitUpResult::getContinuousCount));
        exponent.setMaxContinuousNum(max.isPresent() ? max.get().getContinuousCount() : 0);
        return marketOverviewAccessor.saveOrUpdate(exponent);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public boolean syncDayLines(String date) {
        if (hisDailyLineAccessor.count(new LambdaQueryWrapper<HisDailyLineDO>().eq(HisDailyLineDO::getDate, date)) > 0) {
            return true;
        }
        List<FetchDailyLineResult> results = remoteApiFacade.fetchHisDailyLines(date);
        List<HisDailyLine> dayLines = results.stream().map(e -> {
            HisDailyLine dailyLine = new HisDailyLine();
            dailyLine.setCode(e.getCode());
            dailyLine.setDate(DateUtil.parseDate(e.getDate()));
            dailyLine.setOpen(e.getOpen());
            dailyLine.setClose(e.getPrice());
            dailyLine.setHigh(e.getHigh());
            dailyLine.setLow(e.getLow());
            dailyLine.setVolume(e.getVolume());
            dailyLine.setAmount(e.getAmount());
            dailyLine.setUpPercentage(e.getUpPercentage());
            dailyLine.setUpValue(e.getUpValue());
            return dailyLine;
        }).collect(Collectors.toList());

        return hisDailyLineAccessor.saveBatch(dayLines);
    }
}
